package biz.ws.ws;

import api.common.util.TextUtils;
import biz.ws.WsConfig;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextStoppedEvent;

import javax.annotation.Resource;
import javax.validation.Validation;
import javax.validation.Validator;
import java.net.URL;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @创建人 xiaojiancheng
 * @时间 2020/1/8
 * @描述 ws 服务 不需要单独的线程进行处理
 *
 * 因为功能的缘故 这里考虑到紧凑性，就不拆分出去
 * <li>    整体逻辑代码量不长，如果拆分成多个文件，就比较繁琐
 * <li>    {@link WebsocketService}     具体的功能分发到这里实例里面
 * <li>    这里只是负责 对接 ws 不做任何业务处理，但是
 *          检测是否登录成功
 *          消息的转换
 */
public class Server implements InitializingBean, ApplicationListener<ContextStoppedEvent> {
    @Resource
    UserContainer userContainer;

    @Resource
    MessageConvert messageConvert;

    @Autowired
    WebsocketService websocketService;

    @Resource
    WsConfig wsConfig;

    Validator validator ;

    ChannelFuture future;
    EventLoopGroup boss;
    EventLoopGroup worker;

    AtomicInteger ai = new AtomicInteger(0);

    private void start() {
        //  单线程操作(1io线程+1业务线程)
        boss = new NioEventLoopGroup(wsConfig.getIoThreads());
        worker = new NioEventLoopGroup(wsConfig.getWorkThreads());

        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.option(ChannelOption.SO_BACKLOG, wsConfig.getBacklog());

        bootstrap.group(boss, worker)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline()
                                //  入口
                                .addLast(new EnterHandler())
                                //  心跳处理
                                .addLast(new IdleStateHandler(wsConfig.getBeatTime(), wsConfig.getBeatTime(), wsConfig.getBeatTime() * 2))
                                .addLast(new BeatHandler())
                                //  编码
                                .addLast(new HttpServerCodec())
                                .addLast(new HttpObjectAggregator(wsConfig.getMaxBodySize()))
                                //  登录注册
                                .addLast(new LoginHandler())
                                //  ws
                                .addLast(new WebSocketServerProtocolHandler(wsConfig.getEndPoint(), "", true))
                                //  webSocket 具体报文数据
                                .addLast(new RealDataHandler());
                    }
                });


        future = bootstrap.bind(wsConfig.getPort())
                .addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        websocketService.onStart(future);
                    }
                });
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        validator = Validation.buildDefaultValidatorFactory().getValidator();
        start();
    }

    @Override
    public void onApplicationEvent(ContextStoppedEvent event) {
        if (future != null) {
            future.channel().close();
            worker.shutdownGracefully();
            boss.shutdownGracefully();
        }
    }

    //  心跳处理
    private class BeatHandler extends  ChannelInboundHandlerAdapter{
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            if (evt instanceof IdleStateEvent) {
                if (((IdleStateEvent) evt).state() == IdleState.WRITER_IDLE) {
                    MsgHolder context = new MsgHolder();
                    context.setCtx(ctx);
                    websocketService.onNotBeat(context);
                }
                //  调用原先方法
                super.userEventTriggered(ctx,evt);
                return ;
            }
            super.userEventTriggered(ctx, evt);
        }
    }

    //  登录处理
    private class LoginHandler extends WebSocketServerCompressionHandler {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof HttpRequest) {
                HttpRequest request = (HttpRequest) msg;
                String uri = request.uri();
                URL url = new URL("http://ws"+uri);
                WsModel.LoginInfo loginInfo = ((JSONObject) JSON.toJSON(TextUtils.httpQueryToMap(url.getQuery()))).toJavaObject(WsModel.LoginInfo.class);
                //  参数校验
                validator.validate(loginInfo,WsModel.LoginInfo.class);
                MsgHolder context = new MsgHolder();
                context.setCtx(ctx);
                context.setLoginInfo(loginInfo);
                websocketService.onLogin(context);
                //  一定要修改成这个    否则下面的handler会处理失败
                request.setUri(wsConfig.getEndPoint());
            }
            super.channelRead(ctx, msg);
        }
    }

    //  入口钩子
    private class EnterHandler extends ChannelInboundHandlerAdapter{

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            websocketService.onConnect(new MsgHolder(ctx));
            super.channelActive(ctx);
        }
    }

    //  实际业务的处理
    private class RealDataHandler extends  SimpleChannelInboundHandler<WebSocketFrame> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame frame) throws Exception {
            if (frame instanceof TextWebSocketFrame) {

                String text = ((TextWebSocketFrame) frame).text();
                Message decode = messageConvert.decode(text);

                MsgHolder context = new MsgHolder();
                context.setCtx(ctx);
                context.setMessage(decode);

                //  检测消息是否支持
                if (!WsCst.MsgType.check(decode.getType()) && false) {
                    websocketService.onMessageTypeNotSupport(context);
                    return;
                }

                websocketService.onMessageFromClient(context);
            }
        }

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            userContainer.removeConnection(ctx.channel());
            MsgHolder context = new MsgHolder();
            context.setCtx(ctx);
            websocketService.onUnregister(context);
            super.channelUnregistered(ctx);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            MsgHolder context = new MsgHolder();
            context.setCtx(ctx);
            context.setE(cause);
            websocketService.onException(context);
            super.exceptionCaught(ctx, cause);
        }
    }
}
